/*
 * Copyright (C) 2012  Romain DUBOIS
 * 
 * This file is part of NeMoS - Network and Modular Software
 * 
 * NeMoS is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * NeMoS is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NeMoS.  If not, see <http://www.gnu.org/licenses/>.
 */
package nemos.remote.rsa.message.impl;

import static nemos.remote.rsa.message.MessageConfigurationConstants.FRAMEWORK_EVENT_TOPIC_ROOT;
import static nemos.remote.rsa.message.MessageConfigurationConstants.SERVER;
import static nemos.remote.rsa.message.MessageConfigurationConstants.TRANSPORT_ID_PROPERTY;

import java.io.IOException;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import nemos.remote.rsa.internal.RsaMessage;
import nemos.remote.rsa.message.IServer;
import nemos.remote.rsa.message.data.EFrameworkAction;
import nemos.remote.rsa.message.data.EndpointMessage;
import nemos.remote.rsa.message.data.FrameworkMessage;
import nemos.remote.rsa.message.data.InitMessage;
import nemos.remote.rsa.message.data.InvocationMessage;
import nemos.remote.rsa.message.data.Message;
import nemos.remote.rsa.message.data.ResponseMessage;
import nemos.remote.rsa.message.data.ShutdownMessage;
import nemos.remote.rsa.message.transport.ITransport;
import nemos.remote.rsa.message.transport.LocalTransport;
import nemos.remote.util.RemoteUtil;
import nemos.util.BlockingReference;
import nemos.util.log.NemosLogger;

import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
import org.osgi.util.tracker.ServiceTracker;

/**
 * MessageServer : The message-based server.
 * 
 * <p>
 * This class centralized messages from connected {@link TransportRsa}s and
 * process them.
 * </p>
 * 
 * @author Romain DUBOIS
 */
public class MessageServer implements IServer {

    /** The event admin tracker */
    private ServiceTracker<EventAdmin, EventAdmin> mEventAdmin;

    /** The logger */
    final NemosLogger mLogger;
    /** The last transport number */
    private final AtomicInteger mTransportNb;
    /** The listening threads */
    private final Map<ITransport, Thread> mListeningThreads;
    /** The registered transports with their framework UUIDs */
    private final Map<String, ITransport> mTransports;
    /** The asynchronuous executor */
    private final ExecutorService mAsyncExecutor;
    /** The bundle context */
    private BundleContext mContext;
    /** The server ID */
    private String mId;
    /** The server registration */
    private ServiceRegistration<IServer> mRegistration;
    /** The current auto-exported endpoints */
    private final Set<EndpointDescription> mDescriptions;

    /**
     * Constructor.
     */
    public MessageServer() {
        super();

        mTransportNb = new AtomicInteger(0);
        mListeningThreads = new HashMap<ITransport, Thread>();
        mDescriptions = new HashSet<EndpointDescription>(1);
        mTransports = Collections.synchronizedMap(new HashMap<String, ITransport>());
        mAsyncExecutor = Executors.newCachedThreadPool();
        mLogger = new NemosLogger(MessageServer.class);
    }

    /**
     * Start the server.
     * 
     * @param pContext
     *            The bundle context
     * @param pProps
     *            The server service properties
     */
    public void start(final BundleContext pContext, final Map<String, ?> pProps) {
        mContext = pContext;

        mEventAdmin = new ServiceTracker<EventAdmin, EventAdmin>(pContext, EventAdmin.class, null);
        mEventAdmin.open();

        // Register as a service
        mRegistration = mContext.registerService(IServer.class, this, pProps == null ? null
                : new Hashtable<String, Object>(pProps));
    }

    /**
     * Set the server ID
     * 
     * @param pId
     *            the ID to set
     */
    protected void setId(final String pId) {
        mId = pId;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IServer#stop()
     */
    @Override
    public void stop() {
        mRegistration.unregister();

        // Interrupt listening threads
        final List<Thread> lThreads = new ArrayList<Thread>();
        synchronized (mTransports) {
            for (final Entry<ITransport, Thread> lEntry : mListeningThreads.entrySet()) {
                // Notify of the shutdown
                final ShutdownMessage lMsg = new ShutdownMessage();
                try {
                    lEntry.getKey().send(lMsg);
                } catch (final RemoteException e) {
                    mLogger.error(RsaMessage.SENDING_ERROR, e, lMsg.toString());
                }

                // Then interrupt listening thread
                lEntry.getValue().interrupt();
                lThreads.add(lEntry.getValue());
            }

            mListeningThreads.clear();
            mTransports.clear();
        }

        // Then wait for them
        for (final Thread lThread : lThreads) {
            try {
                lThread.join();
            } catch (final InterruptedException e) {
                mLogger.warn(RsaMessage.INTERRUPTED_THREAD, e);
            }
        }

        mAsyncExecutor.shutdown();
        mEventAdmin.close();
    }

    /**
     * Add a new transport.
     * 
     * @param pTransport
     *            The transport to add
     * @throws RemoteException
     *             If init message sending failed
     */
    public void addTransport(final ITransport pTransport) throws RemoteException {
        final int lNb = mTransportNb.incrementAndGet();
        pTransport.setLabel("S" + lNb); //$NON-NLS-1$
        pTransport.send(new InitMessage("C" + lNb, getConnectedFrameworks())); //$NON-NLS-1$

        new Thread(new Runnable() {
            @Override
            public void run() {
                listen(pTransport);
            }
        }).start();
    }

    /**
     * The transport listening method.
     * 
     * @param pTransport
     *            The transport to listen to
     */
    protected void listen(final ITransport pTransport) {

        // Check and update threads map
        synchronized (mTransports) {
            if (mListeningThreads.get(pTransport) != null) {
                return;
            }
            mListeningThreads.put(pTransport, Thread.currentThread());

            // Notify transport of already registered endpoints
            for (final EndpointDescription lDesc : mDescriptions) {
                final EndpointMessage lMsg = new EndpointMessage(lDesc, true);
                try {
                    pTransport.send(lMsg);
                } catch (final RemoteException e) {
                    mLogger.error(RsaMessage.SENDING_ERROR, e, lMsg.toString());
                }
            }
        }

        // Start listening
        mLogger.debug(RsaMessage.START_LISTENING);
        final ExecutorService lExecutor = Executors.newSingleThreadExecutor();
        try {

            // Listen continuously, until transport or service shutdown
            while (!Thread.interrupted() && pTransport.isActive()) {
                try {
                    final Message lMessage = pTransport.receive(500, TimeUnit.MILLISECONDS);

                    // Message can be null if timeout is reached
                    if (lMessage != null) {

                        // Incoming remote call
                        if (lMessage instanceof InvocationMessage) {
                            lExecutor.execute(new Runnable() {
                                @Override
                                public void run() {
                                    forwardInvocation((InvocationMessage) lMessage, pTransport);
                                }
                            });
                        }

                        // Remote endpoint added or removed
                        else if (lMessage instanceof EndpointMessage) {
                            mAsyncExecutor.execute(new Runnable() {
                                @Override
                                public void run() {
                                    broadcastEndpoint((EndpointMessage) lMessage, pTransport);
                                }
                            });
                        }

                        // Framework supported or no longer supported
                        else if (lMessage instanceof FrameworkMessage) {
                            mAsyncExecutor.execute(new Runnable() {
                                @Override
                                public void run() {
                                    frameworkSupported((FrameworkMessage) lMessage, pTransport);
                                }
                            });
                        }

                        // Unexpected message type
                        else {
                            mLogger.error(RsaMessage.UNEXPECTED_MESSAGE_TYPE, lMessage.getClass()
                                    .getName());
                        }
                    }

                } catch (final RemoteException e) {
                    mLogger.error(RsaMessage.LISTENING_EXCEPTION, e);
                }
            }

        } catch (final InterruptedException e) {
            // Thread interrupted : stop listening
        } finally {
            lExecutor.shutdown();
            synchronized (mTransports) {
                mListeningThreads.remove(pTransport);
            }
            mLogger.debug(RsaMessage.STOP_LISTENING);
            pTransport.close();
        }
    }

    /**
     * A transport add or remove a supported framework.
     * 
     * @param pMessage
     *            The incoming message
     * @param pTransport
     *            The sending transport
     */
    protected void frameworkSupported(final FrameworkMessage pMessage, final ITransport pTransport) {
        if (pMessage.getAction() == EFrameworkAction.JOIN
                || pMessage.getAction() == EFrameworkAction.LEAVE) {
            updateFramework(pMessage.getFrameworkUUID(), pMessage.getAction(), pTransport);
        }
    }

    /**
     * Broadcast an endpoint.
     * 
     * @param pMessage
     *            The message to broadcast
     * @param pTransport
     *            The calling transport
     */
    protected void broadcastEndpoint(final EndpointMessage pMessage, final ITransport pTransport) {
        final EndpointDescription lEndpoint = pMessage.getEndpoint();

        // Check tranport is coherent
        if (lEndpoint != null && pTransport != mTransports.get(lEndpoint.getFrameworkUUID())) {
            mLogger.error(RsaMessage.INVALID_SOURCE_FRAMEWORK, lEndpoint.getFrameworkUUID(),
                    Long.valueOf(pMessage.getId()));
            return;
        }

        // Select valid transports
        final Set<ITransport> lTransports = new HashSet<ITransport>();
        synchronized (mTransports) {
            for (final Entry<String, ITransport> lEntry : mTransports.entrySet()) {
                if (!lEntry.getKey().equals(pMessage.getEndpoint().getFrameworkUUID())) {
                    lTransports.add(lEntry.getValue());
                }
            }
            if (pMessage.isAdded()) {
                mDescriptions.add(lEndpoint);
            } else {
                mDescriptions.remove(lEndpoint);
            }
        }

        // Notify them
        for (final ITransport lTransport : lTransports) {
            try {
                lTransport.send(pMessage);
            } catch (final RemoteException e) {
                mLogger.error(RsaMessage.SENDING_ERROR, e, pMessage.toString());
            }
        }
    }

    /**
     * Forward an invocation message.
     * 
     * @param pMessage
     *            The message to forward
     * @param pTransport
     *            The calling transport
     */
    protected void forwardInvocation(final InvocationMessage pMessage, final ITransport pTransport) {
        final long lIncomingId = pMessage.getId();
        BlockingReference<ResponseMessage> lRef = new BlockingReference<ResponseMessage>();

        final ITransport lTarget = mTransports.get(pMessage.getCalledFrameworkUuid());

        // Calling framework does not match
        if (mTransports.get(pMessage.getCallingFrameworkUuid()) != pTransport) {
            final String lText = mLogger.getTranslator().getMessage(
                    RsaMessage.INVALID_SOURCE_FRAMEWORK, pMessage.getCallingFrameworkUuid(),
                    Long.valueOf(pMessage.getId()));
            lRef.set(new ResponseMessage(0, new RemoteException(lText)));
        }

        // Unknown framework
        else if (lTarget == null) {
            final String lText = mLogger.getTranslator().getMessage(
                    RsaMessage.UNKNOWN_TARGET_FRAMEWORK, pMessage.getCalledFrameworkUuid(),
                    Long.valueOf(pMessage.getId()));
            lRef.set(new ResponseMessage(0, new RemoteException(lText)));
        }

        // Forward call to target framework
        else {
            // Collect response
            try {
                lRef = lTarget.asyncCall(pMessage);
            } catch (final RemoteException e) {
                lRef.set(new ResponseMessage(0, e));
            }
        }

        // Send response back
        final BlockingReference<ResponseMessage> lReference = lRef;
        mAsyncExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    final ResponseMessage lResponse = lReference.get();
                    lResponse.setId(lIncomingId);
                    pTransport.send(lResponse);
                } catch (final RemoteException e) {
                    mLogger.error(RsaMessage.UNABLE_TO_SEND_RESPONSE, e,
                            Long.valueOf(pMessage.getId()));
                } catch (final InterruptedException e) {
                    mLogger.error(RsaMessage.INTERRUPTED_THREAD, e);
                }
            }
        });
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IConnection#getConnectedFrameworks()
     */
    @Override
    public Set<String> getConnectedFrameworks() {
        return Collections.unmodifiableSet(mTransports.keySet());
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IConnection#getId()
     */
    @Override
    public String getId() {
        return mId;
    }

    /**
     * Connect the local framework to the server.
     * 
     * @return The local connection
     * 
     * @throws IOException
     *             If an local transport creation failed
     */
    public ConnectionImpl connect() throws IOException {
        ConnectionImpl lConnection = null;

        // Create a local transport if required
        if (!getConnectedFrameworks().contains(RemoteUtil.getFrameworkUuid())) {
            mLogger.debug(RsaMessage.CREATE_LOCAL_TRANSPORT);

            final LocalTransport lTransport = new LocalTransport(mId);

            // Bound one side to the server
            try {
                addTransport(lTransport.getPeer());
            } catch (final RemoteException e) {
                // No remote exception for local transport
            }

            // And the other to a connection
            lConnection = new ConnectionImpl(lTransport);
            lConnection.start(mContext);
        }
        return lConnection;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IServer#kick(java.lang.String)
     */
    @Override
    public void kick(final String pFramework) {
        if (!RemoteUtil.getFrameworkUuid().equals(pFramework)) {
            updateFramework(pFramework, EFrameworkAction.KICKED, null);

            // Remove framework endpoints
            synchronized (mTransports) {
                for (final EndpointDescription lEndpoint : new HashSet<EndpointDescription>(
                        mDescriptions)) {
                    if (pFramework.equals(lEndpoint.getFrameworkUUID())) {
                        broadcastEndpoint(new EndpointMessage(lEndpoint, false), null);
                    }
                }
            }
        }
    }

    /**
     * Update connected frameworks.
     * 
     * @param pFrameworkUuid
     *            The framework to update
     * @param pAction
     *            The action to take
     * @param pCalling
     *            The calling transport (if applicable)
     */
    private void updateFramework(final String pFrameworkUuid, final EFrameworkAction pAction,
            final ITransport pCalling) {

        boolean lUpdated = false;
        synchronized (mTransports) {
            final ITransport lTransport = mTransports.get(pFrameworkUuid);
            switch (pAction) {
                case JOIN:
                    // New framework : check it is not already supported
                    if (lTransport == null) {
                        mTransports.put(pFrameworkUuid, pCalling);
                        lUpdated = true;
                    } else {
                        mLogger.warn(RsaMessage.FRAMEWORK_ALREADY_SUPPORTED, pFrameworkUuid);
                    }
                    break;

                case LEAVE:
                case KICKED:
                    // Removed framework : check it is supported by the calling
                    // transport
                    if (lTransport == null || (pCalling != null && pCalling != lTransport)) {
                        mLogger.warn(RsaMessage.FRAMEWORK_NOT_SUPPORTED, pFrameworkUuid);
                    } else {
                        lUpdated = true;
                    }
                    break;

                default:
                    // No other value
            }

            if (lUpdated) {
                // Send event to all connected transports
                final FrameworkMessage lMessage = new FrameworkMessage(pFrameworkUuid, pAction);
                for (final ITransport lOtherTransport : mListeningThreads.keySet()) {
                    try {
                        lOtherTransport.send(lMessage);
                    } catch (final RemoteException e) {
                        mLogger.error(RsaMessage.SENDING_ERROR, e, lMessage.toString());
                    }
                }

                // Do not remove too soon, waited to send event
                if (pAction == EFrameworkAction.LEAVE || pAction == EFrameworkAction.KICKED) {
                    mTransports.remove(pFrameworkUuid);

                }
            }
        }

        // Now send event in local framework (if possible and required)
        if (lUpdated) {
            postFrameworkEvent(pFrameworkUuid, pAction);
        }
    }

    /**
     * Post a framework event if EventAdmin is available.
     * 
     * @param pFrameworkUuid
     *            The framework
     * @param pAction
     *            The framework action
     */
    private void postFrameworkEvent(final String pFrameworkUuid, final EFrameworkAction pAction) {
        final EventAdmin lEventAdmin = mEventAdmin.getService();
        if (lEventAdmin != null) {
            final String lTopic = FRAMEWORK_EVENT_TOPIC_ROOT + pAction.name();
            final Map<String, Object> lProperties = new HashMap<String, Object>();
            lProperties.put(SERVER, Boolean.TRUE);
            lProperties.put(TRANSPORT_ID_PROPERTY, getId());
            lProperties.put(Constants.FRAMEWORK_UUID, pFrameworkUuid);

            lEventAdmin.postEvent(new Event(lTopic, lProperties));
        }
    }
}
